package com.caius.xiaohashu.user.relation.biz.consumer;

import com.alibaba.nacos.shaded.com.google.common.util.concurrent.RateLimiter;
import com.caius.framework.common.util.DateUtils;
import com.caius.framework.common.util.JsonUtils;
import com.caius.xiaohashu.user.relation.biz.constant.MQConstants;
import com.caius.xiaohashu.user.relation.biz.constant.RedisKeyConstants;
import com.caius.xiaohashu.user.relation.biz.domain.dataobject.FansDO;
import com.caius.xiaohashu.user.relation.biz.domain.dataobject.FollowingDO;
import com.caius.xiaohashu.user.relation.biz.domain.mapper.FansDOMapper;
import com.caius.xiaohashu.user.relation.biz.domain.mapper.FollowingDOMapper;
import com.caius.xiaohashu.user.relation.biz.enums.FollowUnfollowTypeEnum;
import com.caius.xiaohashu.user.relation.biz.model.dto.CountFollowUnfollowMqDTO;
import com.caius.xiaohashu.user.relation.biz.model.dto.FollowUserMqDTO;
import com.caius.xiaohashu.user.relation.biz.model.dto.UnfollowUserMqDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionTemplate;

import java.time.LocalDateTime;
import java.util.Collections;
import java.util.Objects;

/**
 * @author Caius
 * @description 关注用户消费者
 * @since Created in 2025-05-23
 */
@Component
@RocketMQMessageListener(consumerGroup = "xiaohashu_group_" + MQConstants.TOPIC_FOLLOW_OR_UNFOLLOW,
        topic = MQConstants.TOPIC_FOLLOW_OR_UNFOLLOW,
        consumeMode = ConsumeMode.ORDERLY  // 顺序模式
)
@Slf4j
public class FollowUnfollowConsumer implements RocketMQListener<Message> {

    @Resource
    private FollowingDOMapper followingDOMapper;

    @Resource
    private FansDOMapper fansDOMapper;

    @Resource
    private TransactionTemplate transactionTemplate;

    @Resource
    private RateLimiter rateLimiter;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Override
    public void onMessage(Message message) {
        // 流量削峰：通过获取令牌，如果没有令牌可用，将阻塞，直到获得
        rateLimiter.acquire();

        String bodyJson = new String(message.getBody());

        String tags = message.getTags();

        log.info("==> FollowUnfollowConsumer 消费了信息 {}, tags :{}" , message, tags);

        if (Objects.equals(tags, MQConstants.TAG_FOLLOW)) {
            handleFollowTagMessage(bodyJson);
        } else if (Objects.equals(tags, MQConstants.TAG_UNFOLLOW)){
            handleUnfollowTagMessage(bodyJson);
        }
    }


    /**
     * 取关用户
     * @param bodyJson
     */
    private void handleUnfollowTagMessage(String bodyJson) {
        // json 字符串转换为 DTO 对象
        UnfollowUserMqDTO unfollowUserMqDTO = JsonUtils.parseObject(bodyJson, UnfollowUserMqDTO.class);

        // 判空
        if (Objects.isNull(unfollowUserMqDTO)) return;

        Long userId = unfollowUserMqDTO.getUserId();
        Long unfollowUserId = unfollowUserMqDTO.getUnfollowUserId();
        LocalDateTime createTime = unfollowUserMqDTO.getCreateTime();

        boolean isSuccess = Boolean.TRUE.equals(transactionTemplate.execute(status -> {
            try {
                // 关注表数据删除
                int count = followingDOMapper.deleteByUserIdAndFollowingUserId(userId, unfollowUserId);

                // 粉丝表数据删除
                if (count > 0) {
                    fansDOMapper.deleteByUserIdAndFansUserId(unfollowUserId, userId);
                }
                return true;
            } catch (Exception e) {
                // 标记事件回滚
                status.setRollbackOnly();
                log.error("", e);
            }
            return false;
        }));

        // 更新成功之后 将用户信息从关注用户的粉丝列表删除
        if (isSuccess) {
            String fanRedisKey = RedisKeyConstants.buildUserFanKey(unfollowUserId);
            redisTemplate.opsForZSet().remove(fanRedisKey, userId);

            CountFollowUnfollowMqDTO countFollowUnfollowMqDTO = CountFollowUnfollowMqDTO.builder()
                    .userId(userId)
                    .targetUserId(unfollowUserId)
                    .type(FollowUnfollowTypeEnum.UNFOLLOW.getCode())
                    .build();

            sendMQ(countFollowUnfollowMqDTO);
        }

    }

    /**
     * 关注
     * @param bodyJson
     */
    private void handleFollowTagMessage(String bodyJson) {
        // 转换对象实现
        FollowUserMqDTO followUserMqDTO = JsonUtils.parseObject(bodyJson, FollowUserMqDTO.class);

        if (Objects.isNull(followUserMqDTO)) return;

        // 幂等性：通过联合唯一索引保证

        Long userId = followUserMqDTO.getUserId();
        Long followUserId = followUserMqDTO.getFollowUserId();
        LocalDateTime createTime = followUserMqDTO.getCreateTime();

        // 编程式提交
        boolean isSuccess = Boolean.TRUE.equals(transactionTemplate.execute(status -> {
            try {
                int count = followingDOMapper.insert(FollowingDO.builder()
                        .userId(userId)
                        .followingUserId(followUserId)
                        .createTime(createTime)
                        .build());

                // 关注表更新之后 fans 表更新
                if (count > 0) {
                    fansDOMapper.insert(FansDO.builder()
                            .userId(followUserId)
                            .fansUserId(userId)
                            .createTime(createTime)
                            .build());
                }
                return true;
            } catch (Exception e) {
                status.setRollbackOnly(); //标记为事务回滚
                log.error("", e);
            }
            return false;
        }));

        log.info("## 数据库添加记录结果: {}", isSuccess);

        // 更新 redis 中被关注用户的 ZSet 粉丝列表
        if (isSuccess) {
            // lua 脚本
            DefaultRedisScript<Long> script = new DefaultRedisScript<>();
            script.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/follow_check_and_update_fans_zset.lua")));
            script.setResultType(Long.class);

            long timestamp = DateUtils.localDateTime2Timestamp(createTime);
            // 构建被关注粉丝数列表 redisKey
            String fansRedisKey = RedisKeyConstants.buildUserFanKey(followUserId);
            // 执行 lua 脚本
            Long result = redisTemplate.execute(script, Collections.singletonList(fansRedisKey), userId, timestamp);

            // 发送 MQ 通知计数服务：统计关注数
            // 构建消息体 DTO
            CountFollowUnfollowMqDTO countFollowUnfollowMqDTO = CountFollowUnfollowMqDTO.builder()
                    .userId(userId)
                    .targetUserId(followUserId)
                    .type(FollowUnfollowTypeEnum.FOLLOW.getCode())
                    .build();

            sendMQ(countFollowUnfollowMqDTO);
        }
    }

    /**
     * 发送 MQ 通知计数服务
     * @param countFollowUnfollowMqDTO
     */
    private void sendMQ(CountFollowUnfollowMqDTO countFollowUnfollowMqDTO) {
        org.springframework.messaging.Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(countFollowUnfollowMqDTO))
                .build();

        rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_FOLLOWING, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【计数服务: 关注数】MQ 发送成功， sendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【计数服务: 关注数】MQ 发送异常， " , throwable);
            }
        });

        rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_FANS, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【计数服务：粉丝数】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【计数服务：粉丝数】MQ 发送异常: ", throwable);
            }
        });
    }
}
